package com.yuebook.smartrpc.framework.zookeeper;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.yuebook.smartrpc.framework.helper.IPHelper;
import com.yuebook.smartrpc.framework.helper.PropertyConfigHelper;
import com.yuebook.smartrpc.framework.model.InvokerService;
import com.yuebook.smartrpc.framework.model.ProviderService;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.serialize.SerializableSerializer;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;

/**
 * @author honglei
 * @date 2018/11/18 上午12:01
 */
public class RegisterCenter implements IRegisterCenterProvider, IRegisterCenterInvoker, IRegisterCenterGovernance {

    private static final RegisterCenter registerCenter = new RegisterCenter();

    private static final Map<String, List<ProviderService>> providerServiceMap = Maps.newConcurrentMap();

    private static final Map<String, List<ProviderService>> serviceMetaDataMap = Maps.newConcurrentMap();

    private static String ZK_SERVICE = PropertyConfigHelper.getZkService();
    private static int ZK_SESSION_TIME_OUT = PropertyConfigHelper.getZkConnectionTimeout();
    private static int ZK_CONNECTION_TIME_OUT = PropertyConfigHelper.getZkConnectionTimeout();

    private static String ROOT_PATH = "/config_register";
    public static String PROVIDER_TYPE = "provider";
    public static String INVOKER_TYPE = "consumer";
    private static ZkClient zkClient = null;

    public static RegisterCenter singleton() {
        return registerCenter;
    }

    @Override
    public void registerServices(List<ProviderService> serviceProviders) {
        if (CollectionUtils.isEmpty(serviceProviders)) {
            return;
        }
        synchronized (RegisterCenter.class) {
            for (ProviderService serviceProvider : serviceProviders) {
                String serviceItfKey = serviceProvider.getServiceItf().getName();
                List<ProviderService> providers = providerServiceMap.get(serviceItfKey);
                if (providers == null) {
                    providers = new ArrayList<>();
                }
                providers.add(serviceProvider);
                providerServiceMap.put(serviceItfKey, providers);
            }

            if (zkClient == null) {
                zkClient = new ZkClient(ZK_SERVICE, ZK_SESSION_TIME_OUT, ZK_CONNECTION_TIME_OUT, new SerializableSerializer());
            }

            String APP_KEY = serviceProviders.get(0).getAppKey();
            String ZK_PATH = ROOT_PATH+"/"+APP_KEY;
            if (!zkClient.exists(ZK_PATH)) {
                zkClient.createPersistent(ZK_PATH, true);
            }

            for (Entry<String, List<ProviderService>> entry : providerServiceMap.entrySet()) {
                String groupName = entry.getValue().get(0).getGroupName();
                String serviceNode = entry.getKey();
                String servicePath = ZK_PATH+"/"+groupName+"/"+serviceNode+"/"+PROVIDER_TYPE;
                if (!zkClient.exists(servicePath)) {
                    zkClient.createPersistent(servicePath, true);
                }

                int port = entry.getValue().get(0).getServicePort();
                int weight = entry.getValue().get(0).getWeight();
                int workerThreads = entry.getValue().get(0).getWorkerThreads();
                String ip = IPHelper.localIp();
                String currentServiceIPNode = servicePath+"/"+ip+","+port+","+weight+","+workerThreads+","+groupName;
                if (!zkClient.exists(currentServiceIPNode)) {
                    zkClient.createEphemeral(currentServiceIPNode);
                }

                //注册服务提供者的zk监听事件
                zkClient.subscribeChildChanges(servicePath, new IZkChildListener(){
                    @Override
                    public void handleChildChange(String parentPath, List<String> currentChildren) throws Exception {
                        List<String> serviceIps = new ArrayList<>();
                        if (!CollectionUtils.isEmpty(currentChildren)) {
                            for (String node : currentChildren) {
                                serviceIps.add(node.split(",")[0]);
                            }
                        }
                        refreshActivityService(serviceIps);
                    }
                });

            }
        }
    }

    @Override
    public Map<String, List<ProviderService>> getProviderServiceMap() {
        return providerServiceMap;
    }

    @Override
    public void initServiceMetaData(String remoteAppKey, String groupName) {
        if (MapUtils.isNotEmpty(serviceMetaDataMap)) {
            return;
        }
        Map<String, List<ProviderService>> tempServiceMetaDataMap = Maps.newConcurrentMap();
        synchronized (RegisterCenter.class) {
            if (zkClient == null) {
                zkClient = new ZkClient(ZK_SERVICE, ZK_SESSION_TIME_OUT, ZK_CONNECTION_TIME_OUT, new SerializableSerializer());
            }
            String servicePath = ROOT_PATH+"/"+remoteAppKey+"/"+groupName;
            List<String> serviceList = zkClient.getChildren(servicePath);
            for (String serviceNode : serviceList) {
                String providerPath = servicePath+"/"+serviceNode+"/"+PROVIDER_TYPE;
                List<String> ipInfoList = zkClient.getChildren(providerPath);
                List<ProviderService> providerServiceList = tempServiceMetaDataMap.get(serviceNode);
                if (providerServiceList == null) {
                    providerServiceList = Lists.newArrayList();
                }
                for (String ipInfo : ipInfoList) {
                    String[] providerInfos = ipInfo.split(",");
                    String ip = providerInfos[0];
                    int port = Integer.parseInt(providerInfos[1]);
                    int weight = Integer.parseInt(providerInfos[2]);
                    int workerThreads = Integer.parseInt(providerInfos[3]);
                    String group = providerInfos[4];
                    ProviderService providerService = new ProviderService();
                    try {
                        providerService.setServiceItf(Class.forName(serviceNode));
                    } catch (Exception e) {
                        throw new RuntimeException(e.getMessage());
                    }

                    providerService.setServiceIp(ip);
                    providerService.setServicePort(port);
                    providerService.setWeight(weight);
                    providerService.setWorkerThreads(workerThreads);
                    providerService.setGroupName(group);
                    providerServiceList.add(providerService);
                    tempServiceMetaDataMap.put(serviceNode, providerServiceList);
                }
            }
            serviceMetaDataMap.clear();
            serviceMetaDataMap.putAll(tempServiceMetaDataMap);
        }
    }

    @Override
    public Map<String, List<ProviderService>> getServiceMetaData() {
        return serviceMetaDataMap;
    }

    @Override
    public void registerInvokerService(InvokerService invokerService) {
        if (invokerService == null) {
            return;
        }
        synchronized (RegisterCenter.class) {
            if (zkClient == null) {
                zkClient = new ZkClient(ZK_SERVICE, ZK_SESSION_TIME_OUT, ZK_CONNECTION_TIME_OUT, new SerializableSerializer());
            }
            String invokeName = invokerService.getServiceItf().getName();
            String invokerPath = ROOT_PATH+"/"+invokerService.getRemoteAppKey()+"/"+invokerService.getGroup()+"/"+invokeName+"/"+INVOKER_TYPE;
            if (!zkClient.exists(invokerPath)) {
                zkClient.createPersistent(invokerPath, true);
            }
            String ip = IPHelper.localIp();
            String currentInvokerIpPath = invokerPath+"/"+ip;
            if (!zkClient.exists(currentInvokerIpPath)) {
                zkClient.createEphemeral(currentInvokerIpPath);
            }
        }
    }

    //刷新活跃提供者服务信息
    private void refreshActivityService(List<String> serviceIpList) {
        if (CollectionUtils.isEmpty(serviceIpList)) {
            providerServiceMap.clear();
            return;
        }
        Map<String, List<ProviderService>> currentProviderServiceMap = new ConcurrentHashMap<>();
        for (Entry<String, List<ProviderService>> entry : providerServiceMap.entrySet()) {
            String key = entry.getKey();
            List<ProviderService> currentProviderServices = currentProviderServiceMap.get(key);
            if (currentProviderServices == null) {
                currentProviderServices = new ArrayList<>();
            }
            if (serviceIpList.contains(entry.getValue().get(0).getServiceIp())) {
                currentProviderServices.addAll(entry.getValue());
            }
            currentProviderServiceMap.put(key, currentProviderServices);
        }
        providerServiceMap.clear();
        providerServiceMap.putAll(currentProviderServiceMap);
    }

    @Override
    public Pair<List<ProviderService>, List<InvokerService>> queryProvidersAndInvokers(String serviceName,
        String appKey) {
        List<ProviderService> providerServices = Lists.newArrayList();
        List<InvokerService> invokerServices = Lists.newArrayList();
        if (zkClient == null) {
            synchronized (RegisterCenter.class) {
                if (zkClient == null) {
                    zkClient = new ZkClient(ZK_SERVICE, ZK_SESSION_TIME_OUT, ZK_CONNECTION_TIME_OUT, new SerializableSerializer());
                }
            }
        }
        String app_path = ROOT_PATH+"/"+appKey;
        List<String> groups = zkClient.getChildren(app_path);
        if (CollectionUtils.isEmpty(groups)) {
            return Pair.of(providerServices, invokerServices);
        }
        for (String group : groups) {
            String group_path = app_path+"/"+group;
            List<String> services = zkClient.getChildren(group_path);
            if (CollectionUtils.isEmpty(services)) {
                continue;
            }
            for (String service : services) {
                String service_path = group_path+"/"+service;
                List<String> serviceTypes = zkClient.getChildren(service_path);
                if (CollectionUtils.isEmpty(serviceTypes)) {
                    continue;
                }
                for (String type : serviceTypes) {
                    if (StringUtils.equals(type, PROVIDER_TYPE)) {
                        String provider_path = service_path+"/"+PROVIDER_TYPE;
                        List<String> ipInfos = zkClient.getChildren(provider_path);
                        if (CollectionUtils.isEmpty(ipInfos)) {
                            continue;
                        }
                        for (String ipInfo : ipInfos) {
                            String[] ipMess = ipInfo.split(",");
                            ProviderService providerService = new ProviderService();
                            providerService.setServiceIp(ipMess[0]);
                            providerService.setServicePort(Integer.parseInt(ipMess[1]));
                            providerService.setWeight(Integer.parseInt(ipMess[2]));
                            providerService.setWorkerThreads(Integer.parseInt(ipMess[3]));
                            providerService.setGroupName(group);
                            providerServices.add(providerService);
                        }
                     } else {
                        String invoker_path = service_path+"/"+INVOKER_TYPE;
                        List<String> ipInfos = zkClient.getChildren(invoker_path);
                        if (CollectionUtils.isEmpty(ipInfos)) {
                            continue;
                        }
                        for (String ipInfo : ipInfos) {
                            InvokerService invokerService = new InvokerService();
                            invokerService.setInvokeIp(ipInfo);
                            invokerService.setRemoteAppKey(appKey);
                            invokerService.setGroup(group);
                            invokerServices.add(invokerService);
                        }
                    }
                }
            }
        }
        return Pair.of(providerServices, invokerServices);
    }
}
